package com.stm.bi.repository;


import com.alibaba.fastjson.JSON;
import com.stm.bi.core.util.DirPathUtil;
import com.stm.framework.context.TraceContext;
import com.stm.framework.core.util.*;
import com.stm.bi.jdbc.DataSourceEnum;
import com.stm.bi.jdbc.DorisJdbcUtils;
import com.stm.bi.repository.dataobject.DorisLoadRecordDO;
import com.stm.framework.dbcp.DataSourceConfig;
import jakarta.annotation.Resource;

import org.apache.commons.codec.binary.Base64;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.FileEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.DefaultRedirectStrategy;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpHeaders;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;



@Component
public class DorisLoadRepository {
    private static final Logger log = LoggerFactory.getLogger(DorisLoadRepository.class);
    private final static String COLUMN_SEPARATOR = "/|";
    @Value("${spring.datasource.jdbcUrl}")
    private String dorisUrl;
    @Value("${spring.datasource.username}")
    private String username;
    @Value("${spring.datasource.password}")
    private String password;
    @Resource
    private DirPathUtil dirPathUtil;
    @Resource
    private DorisLoadRecordRepository dorisLoadRecordRepository;
    @Resource(name = DataSourceConfig.DEFAULT_DATASOURCE)
    private DataSource defaultDataSource;

    private final static HttpClientBuilder httpClientBuilder = HttpClients
            .custom()
            .setRedirectStrategy(new DefaultRedirectStrategy() {
                @Override
                protected boolean isRedirectable(String method) {
                    // 如果连接目标是 FE，则需要处理 307 redirect。
                    return true;
                }
            });

    public void putLoad(String tableName, List<Map> dataList){
        String schema = (String) ClassUtil.getFieldValueByName( defaultDataSource,"schema");
        putLoad(schema,tableName,dataList);
    }

    public void putLoad(DataSourceEnum dataSourceEnum, String tableName, List<Map> dataList){
        JdbcTemplate jdbcTemplate = new JdbcTemplate(SpringContextUtil.getBean(dataSourceEnum.getDataSource()));
        DataSource dataSource = jdbcTemplate.getDataSource();
        String schema = (String) ClassUtil.getFieldValueByName( dataSource,"schema");
        putLoad(schema,tableName,dataList);
    }
    public void putLoad(String schema, String tableName, List<Map> dataList){
        long startTime = System.currentTimeMillis();
        log.info("开始导入数据到Doris,schema:{},tableName:{},数据条数:{}",schema,tableName,dataList.size());
        JdbcTemplate jdbcTemplate = new JdbcTemplate(SpringContextUtil.getBean(DataSourceEnum.ODS.getDataSource()));
        List<Map<String,Object>> columnList = DorisJdbcUtils.getColumnList(jdbcTemplate,schema,tableName);
        List<String> columnNames = columnList.stream().map(map -> (String)map.get("column_name")).collect(Collectors.toList());
        String csvFilePath = dirPathUtil.getBasePath()+"/"+ UUID.genUUID32() +".csv";
        File csvFile = new File(csvFilePath);
        try {
            FileWriter writer = new FileWriter(csvFile);
            for(Map rowMap : dataList){
                String rowDatas = null;
                for(Map<String,Object> columnMap : columnList){
                    String dataType = (String) columnMap.get("data_type");
                    String columnName = (String) columnMap.get("column_name");
                    Object rowValue = rowMap.get(columnName);
                    if(rowValue == null){
                        String humpColumnName = HumpUtil.underline2hump(columnName);
                        rowValue = rowMap.get(humpColumnName);
                    }
                    Object val = convert(dataType,rowValue);
                    if(rowDatas == null){
                        rowDatas = val+"";
                    }else{
                        rowDatas += COLUMN_SEPARATOR+val;
                    }
                }
                rowDatas = rowDatas.replace("\n","");
                writer.write(rowDatas+"\n");
            }
            writer.flush();
            // 更多行数据...
            writer.close();
            String columns = columnNames.stream().collect(Collectors.joining(","));
            putLoad(schema,tableName,csvFile,columns,COLUMN_SEPARATOR);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }finally {
            log.info("导入数据到Doris完成,耗时:{}ms",System.currentTimeMillis()-startTime);
            csvFile.delete();
        }
    }


    public void putLoad(String schema, String tableName,File file,String columns,String separator) {
        String host = dorisUrl.split(":")[2].replace("//","");
        int port = Integer.parseInt(dorisUrl.split(":")[3].split("/")[0]) - 1000;
        try (CloseableHttpClient client = httpClientBuilder.build()) {
            HttpPut put = new HttpPut("http://"+host+":"+port+"/api/"+schema+"/"+tableName+"/_stream_load");
            put.setHeader(HttpHeaders.EXPECT, "100-continue");
            put.setHeader(HttpHeaders.AUTHORIZATION, basicAuthHeader(username, password));
            put.setHeader("Expect","100-continue");
            put.setHeader("column_separator",separator);
            put.setHeader("columns",columns);
            // 设置错误行过滤比例
            put.setHeader("max_filter_ratio","0.1");
            // 设置导入文件。
            // 这里也可以使用 StringEntity 来传输任意数据。
            FileEntity entity = new FileEntity(file);
            put.setEntity(entity);
            try (CloseableHttpResponse response = client.execute(put)) {
                String loadResult = "";
                if (response.getEntity() != null) {
                    loadResult = EntityUtils.toString(response.getEntity());
                }
                final int statusCode = response.getStatusLine().getStatusCode();
                if (statusCode != 200) {
                    throw new IOException(String.format("导入失败. status: %s load result: %s", statusCode, loadResult));
                }
                DorisLoadRecordDO dorisLoadRecord = JSON.parseObject(loadResult, DorisLoadRecordDO.class);
                dorisLoadRecord.setLoadTime(LocalDateTime.now());
                dorisLoadRecord.setTraceId(TraceContext.getId());
                dorisLoadRecord.setTableName(tableName);
                dorisLoadRecord.setSchemaName(schema);
                dorisLoadRecordRepository.create(dorisLoadRecord);
                log.info("Get load result: {}", loadResult);
                if(dorisLoadRecord.getNumberFilteredRows() >0){
                    dorisLoadRecord.setLoadTime(LocalDateTime.now());
                    dorisLoadRecord.setTraceId(TraceContext.getId());
                    dorisLoadRecord.setTableName(tableName);
                    dorisLoadRecord.setSchemaName(schema);
                    dorisLoadRecord.setStatus("Fail");
                    dorisLoadRecordRepository.create(dorisLoadRecord);
                    throw new RuntimeException(dorisLoadRecord.getMessage());
                }
            }
        }catch (Exception e) {
            log.error("上传文件：{}到Doris报错{}",file.getName(),e);
            DorisLoadRecordDO dorisLoadRecord = new DorisLoadRecordDO();
            dorisLoadRecord.setLoadTime(LocalDateTime.now());
            dorisLoadRecord.setTraceId(TraceContext.getId());
            dorisLoadRecord.setTableName(tableName);
            dorisLoadRecord.setSchemaName(schema);
            dorisLoadRecord.setStatus("Fail");
            dorisLoadRecord.setMessage(e.getMessage());
            dorisLoadRecordRepository.create(dorisLoadRecord);
            throw new RuntimeException(e);
        }
    }

    private String basicAuthHeader(String username, String password) {
        final String tobeEncode = username + ":" + password;
        //log.info("用户名：密码{}",tobeEncode);
        byte[] encoded = Base64.encodeBase64(tobeEncode.getBytes(StandardCharsets.UTF_8));
        return "Basic " + new String(encoded);
    }

    /**
     * 根据数据库数据类型转换值
     *
     * @param dbDataType 数据库中的数据类型
     * @param value 待插入的值
     * @return 转换后的值，适合数据库插入
     */
    public static Object convert(String dbDataType,Object value){
        // 如果值为null，则直接返回null
        if(value == null){
            return "";
        }
        if(value instanceof Boolean){
            return ((Boolean)value)?1:0;
        }else if(DorisJdbcUtils.DATA_TYPE_STRING_ARRAY.contains(dbDataType)){
            return value;
        }else if(DorisJdbcUtils.DATA_TYPE_NUMBER_ARRAY.contains(dbDataType)){
            return value;
        }else if(DorisJdbcUtils.DATA_TYPE_DATE_TIME_ARRAY.contains(dbDataType)){
            if(value instanceof LocalDateTime){
                return ((LocalDateTime)value).format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            }else if(value instanceof LocalDate){
                return ((LocalDate)value).format(DateTimeFormatter.ofPattern("yyyy-MM-dd"))+" 00:00:00";
            }else if(value instanceof Date){
                return (DateUtils.parseDateToStr((Date)value, "yyyy-MM-dd HH:mm:ss"));
            }else if(value instanceof Long){
                return (DateUtils.parseDateToStr(new Date((Long)value), "yyyy-MM-dd HH:mm:ss"));
            }else {
                return value;
            }
        }else if(DorisJdbcUtils.DATA_TYPE_DATE_ARRAY.contains(dbDataType)){
            if(value instanceof LocalDateTime){
                return ((LocalDateTime)value).format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
            }else if(value instanceof LocalDate){
                return ((LocalDate)value).format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
            }else if(value instanceof Date){
                return (DateUtils.parseDateToStr((Date)value, "yyyy-MM-dd"));
            }else if(value instanceof Long){
                return (DateUtils.parseDateToStr(new Date((Long)value), "yyyy-MM-dd"));
            }else{
                return value;
            }
        }else{
            return value;
        }
    }

    public static void main(String[] args) throws Exception{
        DorisLoadRepository loader = new DorisLoadRepository();
        File file = new File("D://doris.csv");

        String columns = "dept_id,dept_name,date,year_month_str,year_id,wide_group_id,prod_id,prod_name,is_inseason,inseason_name,opening_qty,ending_qty,warehousing_qty,outofstock_qty,id,color_id,color_name,size_id,size_name,style_type,ms";
        loader.putLoad("dmp_ads","ads_inv_order_wide",file,columns,"\u0001");

    }
}